/*
 * Copyright 2009-2013 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.uci.ics.hyracks.dataflow.std.join;

import java.io.DataOutput;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.value.INullWriter;
import edu.uci.ics.hyracks.api.dataflow.value.IPredicateEvaluator;
import edu.uci.ics.hyracks.api.dataflow.value.ITuplePairComparator;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.io.RunFileReader;
import edu.uci.ics.hyracks.dataflow.common.io.RunFileWriter;

public class NestedLoopJoin {
    private final FrameTupleAccessor accessorInner;
    private final FrameTupleAccessor accessorOuter;
    private final FrameTupleAppender appender;
    private final ITuplePairComparator tpComparator;
    private final ByteBuffer outBuffer;
    private final ByteBuffer innerBuffer;
    private final List<ByteBuffer> outBuffers;
    private final int memSize;
    private final IHyracksTaskContext ctx;
    private RunFileReader runFileReader;
    private int currentMemSize = 0;
    private final RunFileWriter runFileWriter;
    private final boolean isLeftOuter;
    private final ArrayTupleBuilder nullTupleBuilder;
    private final IPredicateEvaluator predEvaluator;
    private boolean isReversed;		//Added for handling correct calling for predicate-evaluator upon recursive calls (in OptimizedHybridHashJoin) that cause role-reversal

    
    public NestedLoopJoin(IHyracksTaskContext ctx, FrameTupleAccessor accessor0, FrameTupleAccessor accessor1,
            ITuplePairComparator comparators, int memSize, IPredicateEvaluator predEval, boolean isLeftOuter, INullWriter[] nullWriters1)
            throws HyracksDataException {
        this.accessorInner = accessor1;
        this.accessorOuter = accessor0;
        this.appender = new FrameTupleAppender(ctx.getFrameSize());
        this.tpComparator = comparators;
        this.outBuffer = ctx.allocateFrame();
        this.innerBuffer = ctx.allocateFrame();
        this.appender.reset(outBuffer, true);
        this.outBuffers = new ArrayList<ByteBuffer>();
        this.memSize = memSize;
        this.predEvaluator = predEval;
        this.isReversed = false;
        this.ctx = ctx;

        this.isLeftOuter = isLeftOuter;
        if (isLeftOuter) {
            int innerFieldCount = accessorInner.getFieldCount();
            nullTupleBuilder = new ArrayTupleBuilder(innerFieldCount);
            DataOutput out = nullTupleBuilder.getDataOutput();
            for (int i = 0; i < innerFieldCount; i++) {
                nullWriters1[i].writeNull(out);
                nullTupleBuilder.addFieldEndOffset();
            }
        } else {
            nullTupleBuilder = null;
        }

        FileReference file = ctx.getJobletContext().createManagedWorkspaceFile(
                this.getClass().getSimpleName() + this.toString());
        runFileWriter = new RunFileWriter(file, ctx.getIOManager());
        runFileWriter.open();
    }

    public void cache(ByteBuffer buffer) throws HyracksDataException {
        runFileWriter.nextFrame(buffer);
    }

    public void join(ByteBuffer outerBuffer, IFrameWriter writer) throws HyracksDataException {
        if (outBuffers.size() < memSize - 3) {
            createAndCopyFrame(outerBuffer);
            return;
        }
        if (currentMemSize < memSize - 3) {
            reloadFrame(outerBuffer);
            return;
        }
        runFileReader = runFileWriter.createReader();
        runFileReader.open();
        while (runFileReader.nextFrame(innerBuffer)) {
            for (ByteBuffer outBuffer : outBuffers) {
                blockJoin(outBuffer, innerBuffer, writer);
            }
        }
        runFileReader.close();
        currentMemSize = 0;
        reloadFrame(outerBuffer);
    }

    private void createAndCopyFrame(ByteBuffer outerBuffer) throws HyracksDataException {
        ByteBuffer outerBufferCopy = ctx.allocateFrame();
        FrameUtils.copy(outerBuffer, outerBufferCopy);
        outBuffers.add(outerBufferCopy);
        currentMemSize++;
    }

    private void reloadFrame(ByteBuffer outerBuffer) {
        outBuffers.get(currentMemSize).clear();
        FrameUtils.copy(outerBuffer, outBuffers.get(currentMemSize));
        currentMemSize++;
    }

    private void blockJoin(ByteBuffer outerBuffer, ByteBuffer innerBuffer, IFrameWriter writer)
            throws HyracksDataException {
        accessorOuter.reset(outerBuffer);
        accessorInner.reset(innerBuffer);
        int tupleCount0 = accessorOuter.getTupleCount();
        int tupleCount1 = accessorInner.getTupleCount();

        for (int i = 0; i < tupleCount0; ++i) {
            boolean matchFound = false;
            for (int j = 0; j < tupleCount1; ++j) {
                int c = compare(accessorOuter, i, accessorInner, j);
                boolean prdEval = evaluatePredicate(i, j);
                if (c == 0 && prdEval) {
                	matchFound = true;
                	appendToResults(i, j, writer);
                }
            }

            if (!matchFound && isLeftOuter) {
                final int[] ntFieldEndOffsets = nullTupleBuilder.getFieldEndOffsets();
                final byte[] ntByteArray = nullTupleBuilder.getByteArray();
                final int ntSize = nullTupleBuilder.getSize();
                if (!appender.appendConcat(accessorOuter, i, ntFieldEndOffsets, ntByteArray, 0, ntSize)) {
                    flushFrame(outBuffer, writer);
                    appender.reset(outBuffer, true);
                    if (!appender.appendConcat(accessorOuter, i, ntFieldEndOffsets, ntByteArray, 0, ntSize)) {
                        int tSize = accessorOuter.getTupleEndOffset(i) - accessorOuter.getTupleStartOffset(i) + ntSize;
                        throw new HyracksDataException("Record size (" + tSize + ") larger than frame size ("
                                + appender.getBuffer().capacity() + ")");
                    }
                }
            }
        }
    }
    
    private boolean evaluatePredicate(int tIx1, int tIx2){
    	if(isReversed){		//Role Reversal Optimization is triggered
    		return ( (predEvaluator == null) || predEvaluator.evaluate(accessorInner, tIx2, accessorOuter, tIx1) );
    	}
    	else {
    		return ( (predEvaluator == null) || predEvaluator.evaluate(accessorOuter, tIx1, accessorInner, tIx2) );
    	}
    }
    
    private void appendToResults(int outerTupleId, int innerTupleId, IFrameWriter writer) throws HyracksDataException {
        if (!isReversed) {
            appendResultToFrame(accessorOuter, outerTupleId, accessorInner, innerTupleId, writer);
        } else {
            //Role Reversal Optimization is triggered
            appendResultToFrame(accessorInner, innerTupleId, accessorOuter, outerTupleId, writer);
        }
    }

    private void appendResultToFrame(FrameTupleAccessor accessor1, int tupleId1, FrameTupleAccessor accessor2,
            int tupleId2, IFrameWriter writer) throws HyracksDataException {
        if (!appender.appendConcat(accessor1, tupleId1, accessor2, tupleId2)) {
            flushFrame(outBuffer, writer);
            appender.reset(outBuffer, true);
            if (!appender.appendConcat(accessor1, tupleId1, accessor2, tupleId2)) {
                int tSize = accessor1.getTupleEndOffset(tupleId1) - accessor1.getTupleStartOffset(tupleId1)
                        + accessor2.getTupleEndOffset(tupleId2) - accessor2.getTupleStartOffset(tupleId2);
                throw new HyracksDataException("Record size (" + tSize + ") larger than frame size ("
                        + appender.getBuffer().capacity() + ")");
            }
        }
    }
    
    public void closeCache() throws HyracksDataException {
        if (runFileWriter != null) {
            runFileWriter.close();
        }
    }

    public void closeJoin(IFrameWriter writer) throws HyracksDataException {
        runFileReader = runFileWriter.createReader();
        runFileReader.open();
        while (runFileReader.nextFrame(innerBuffer)) {
            for (int i = 0; i < currentMemSize; i++) {
                blockJoin(outBuffers.get(i), innerBuffer, writer);
            }
        }
        runFileReader.close();
        outBuffers.clear();
        currentMemSize = 0;

        if (appender.getTupleCount() > 0) {
            flushFrame(outBuffer, writer);
        }
    }

    private void flushFrame(ByteBuffer buffer, IFrameWriter writer) throws HyracksDataException {
        buffer.position(0);
        buffer.limit(buffer.capacity());
        writer.nextFrame(buffer);
        buffer.position(0);
        buffer.limit(buffer.capacity());
    }

    private int compare(FrameTupleAccessor accessor0, int tIndex0, FrameTupleAccessor accessor1, int tIndex1)
            throws HyracksDataException {
        int c = tpComparator.compare(accessor0, tIndex0, accessor1, tIndex1);
        if (c != 0) {
            return c;
        }
        return 0;
    }
    
    public void setIsReversed(boolean b){
    	this.isReversed = b;
    }
}
